package com.qyer.commons.concurrent.assemblingline;

import static com.qyer.commons.concurrent.assemblingline.ProcessMode.BULK;
import static com.qyer.commons.utils.CommonUtils.timeCost;

import com.qyer.commons.CheckStatus;
import com.qyer.commons.concurrent.ArrayListBulk;
import com.qyer.commons.concurrent.GenericLinkedBlockingQueue;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.commons.lang3.builder.ToStringStyle;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.util.concurrent.CountDownLatch;

/**
 * User: Z J Wu Date: 2016/06/02 Time: 17:53 Package: com.qyer.commons.concurrent
 */
public class Summarizer<T> implements Runnable, Closeable, CheckStatus {

  private static final Logger LOGGER = LoggerFactory.getLogger(Summarizer.class);

  private String name;
  private GenericLinkedBlockingQueue<ArrayListBulk<T>> in;
  private CountDownLatch signal;
  private AbstractHandler<ArrayListBulk<T>, ArrayListBulk<T>> handler;

  private StatisticContext context;
  private AssemblingLineStatus status;

  public Summarizer(String name, CountDownLatch signal,
                    AbstractHandler<ArrayListBulk<T>, ArrayListBulk<T>> handler,
                    StatisticContext context) {
    this(name, null, signal, handler, context);
  }

  public Summarizer(String name, GenericLinkedBlockingQueue<ArrayListBulk<T>> in,
                    CountDownLatch signal,
                    AbstractHandler<ArrayListBulk<T>, ArrayListBulk<T>> handler,
                    StatisticContext context) {
    this.name = name;
    this.in = in;
    this.signal = signal;
    this.handler = handler;
    this.context = context;
    this.status = check();
  }

  @Override
  public void run() {
    try {
      while (true) {
        ArrayListBulk<T> bulk = in.take();
        if (bulk.isPill()) {
          break;
        }
        long t1 = System.currentTimeMillis();
        try {
          handler.handle(bulk);
          context.reportSuccess(name, BULK);
        } catch (HandleException e) {
          LOGGER.warn("Handler cannot handle this msg.", e);
          context.reportFailure(name, BULK);
        } finally {
          context.reportTimeCost(name, timeCost(t1), BULK);
        }
      }
      LOGGER.info("Thread stop.");
    } catch (InterruptedException e) {
      LOGGER.error("Query worker is interrupted", e);
    } catch (Exception e) {
      LOGGER.error("Query worker aborted with error.", e);
    } finally {
      this.signal.countDown();
    }
  }

  public String getName() {
    return name;
  }

  public void assembleConveyors(GenericLinkedBlockingQueue<ArrayListBulk<T>>... conveyors) {
    if (ArrayUtils.isNotEmpty(conveyors)) {
      this.in = conveyors[0];
      this.status = check();
    }
  }

  @Override
  public void close() {
    if (handler instanceof Closeable) {
      IOUtils.closeQuietly((Closeable) handler);
    }
  }

  @Override
  public AssemblingLineStatus check() {
    if (StringUtils.isEmpty(
      this.name) || this.in == null || this.signal == null || this.handler == null || this
      .context == null) {
      return AssemblingLineStatus.INITED;
    } else {
      return AssemblingLineStatus.STANDBY;
    }
  }

  @Override
  public String toString() {
    return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE).append("name", name)
                                                                      .append("in", in)
                                                                      .append("status", status)
                                                                      .toString();
  }
}
